package com.ydzt.fanka.server.space.socketreceiver;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Date;
import java.util.List;

import org.apache.log4j.Logger;
import org.dom4j.Element;

import com.ydzt.fanka.server.domain.CommunicationNode;
import com.ydzt.fanka.server.init.Ask;
import com.ydzt.fanka.server.init.InitData;
import com.ydzt.fanka.server.init.SynchronizedVariable;
import com.ydzt.fanka.server.log.LoggerFanka;
import com.ydzt.fanka.server.persistence.BufferStore;
import com.ydzt.fanka.server.utils.ByteConvert;
import com.ydzt.fanka.server.utils.ConfigFileUtil;
import com.ydzt.fanka.server.utils.Convert;
import com.ydzt.fanka.server.utils.DBHelper;
import com.ydzt.fanka.server.utils.DataTranslateException;
import com.ydzt.fanka.server.utils.FileUtil;
import com.ydzt.fanka.server.utils.NetUtil;
import com.ydzt.fanka.server.utils.PackageHelper;
import com.ydzt.fanka.server.zwcl.ZwclEndDayService;

/*
 * 此线程接受来自socket的报文
 * */
public class SocketReceiverServer extends Thread {
	String tmpMap;
	List<CommunicationNode> broadcastList = null;
	static Logger logger = Logger.getLogger(SocketReceiverServer.class
			.getName());
	List<Element> drList = null;
	ServerSocket serverSocket;
	Socket socket = null;
	DataInputStream in = null;
	DataOutputStream out = null;
	String paramType = null;
	File fTmpdata_centre;
	File fTmpdata_centre_onlyread;
	File fTmpdata_dr;
	String supplier_id;
	SynchronizedVariable synValue;

	byte[] pkg2002_check;
	byte[] pkg3001;
	byte[] head;
	byte[] pkg;
	byte[] msgTypeByte;
	byte[] sender;
	byte[] receiver;
	byte[] msgMgrType;
	byte[] seq;
	byte[] pkgno;
	byte[] flag;
	byte[] recordcount;
	byte[] agm;
	byte[] portByte;
	byte[] activityStmtDayByteArray;
	short port;
	byte response;

	String strSupplierID;
	String strSUpplierIP;
	String strReceiverIP;
	String strReceiverID;

	byte[] byteArrayStmtDay;
	String strDrStmtDay;
	String strCentreStmtDay;
	String strNextStmtDay;
	Date DrStmtDay;
	Date CentreStmtDay;
	long days;
	String activityStmtDay;
	String dr_tmp_path_buffer;
	String dr_tmp_path_onlyread;

	int headLength = 23;

	public SocketReceiverServer(Socket socket) throws IOException {
		this.socket = socket;
		this.in = new DataInputStream(socket.getInputStream());
		this.out = new DataOutputStream(socket.getOutputStream());
		logger.info("初始化SocketReceiverServer");
	};

	public SocketReceiverServer(ServerSocket serverSocket) {
		this.serverSocket = serverSocket;
		logger.info("初始化SocketReceiverServer");
	}

	public SocketReceiverServer() {
		logger.info("初始化SocketReceiverServer");
	}

	public void run() {
		Socket socket = null;
		DataInputStream in = null;
		DataOutputStream out = null;
		logger.info("启动SocketReceiverServer。。。");

		try {
			int port = ConfigFileUtil.getOpenPort();
			serverSocket = new ServerSocket(port);
			logger.info("建立SockerServer");
		} catch (IOException e) {
			logger.info(e.getMessage());
			System.exit(1);
		}

		while (true) {
			try {

				socket = serverSocket.accept();
				in = new DataInputStream(socket.getInputStream());
				// 获取报文长度
				byte[] lenByte = new byte[2];
				in.read(lenByte);

				try {
					// 解析报文长度
					int len = ByteConvert.byteToShort(lenByte);
					logger.info("将从网络读取长度为:" + len + "的包");
					if (len > 23) {
						this.pkg = new byte[len];
						// 读取长度完整的报文
						in.read(this.pkg, 0, len);
						logger.info(LoggerFanka.dumpBin(this.pkg));
						// logger.info("读取的包为:");
						// logger.info(LoggerFanka.dumpBin(this.pkg));
						// 对包头进行解析
						response = this.getResponse(pkg);
						logger.info("");
						logger.info(response);
						out = new DataOutputStream(socket.getOutputStream());
						out.write(response);
						// 包头解析正确
						if (response == 0x00) {
							// 判断是否食堂收到日终报文(2001报文)
							if (paramType.equals("2001")
									&& !InitData.nodeType.equals("00")) {// 日终报文
																			// 放入endflag去处理
																			// 食堂收到2001报文
								logger.info("食堂收到2001报文，将进行日终操作");
								logger.info("将2001报文写入endflag文件夹");
								if (0 == BufferStore
										.writePackageIntoEndFlag(pkg)) {
									logger.info("写入成功");
									ZwclEndDayService zwclEndDayService = new ZwclEndDayService();
									// 调用日终处理方法
									zwclEndDayService.EndDay();
								}
							}//食堂收到解锁进程的报文
							else if(paramType.equals("2003")
									&& !InitData.nodeType.equals("00")){
								logger.info("释放消费者线程(SocketConsumer) unlock");
								SynchronizedVariable.getInstance().setBuffLock(false);
							}// 食堂收到日终报文
							else if (paramType.equals("2002")
									&& !InitData.nodeType.equals("00")) {
								logger.info("食堂收到日终命令");
								
								activityStmtDayByteArray = new byte[4];
								System.arraycopy(this.pkg, 23,
										activityStmtDayByteArray, 0, 4);
								activityStmtDay = ByteConvert
										.bcd2Str(activityStmtDayByteArray);
								logger.info("解析得到的下待生效结算日期为" + activityStmtDay);
								// 判断食堂是否需要日切
								String DrStmtDay = DBHelper.getStmtDay();
								Date dDrStmtDay = Convert.getDateByString(
										DrStmtDay, "yyyyMMdd");
								Date dCentreStmtDay = Convert.getDateByString(
										activityStmtDay, "yyyyMMdd");
								// 比较两个日期
								long timespan = dCentreStmtDay.getTime()
										- dDrStmtDay.getTime();
								days = timespan / InitData.DAY;
								if (days > 0) {
									// 需要进行日切
									logger.info("锁住消费者线程(SocketConsumer) lock");
									SynchronizedVariable.getInstance().setBuffLock(true);
									// 检查数据是否处理完成
									logger.info("开始检查消息队列是否处理完成");
									fTmpdata_dr = new File(InitData.tmpdata);
									boolean isCompleta = true;
									while (isCompleta) {
										String strNextDay = Convert
												.getSpecifiedDayAfter(DrStmtDay);
										long countBuffer = FileUtil
												.getlist(fTmpdata_dr);
										logger.info("在tmpdata文件夹中发现  "
												+ countBuffer + " 条数据未处理");
										if (countBuffer == 0) {// 表示tmpdata文件夹中无文件
																// 文件已经被处理完
											logger.info("tmpdata文件夹中没有发现未处理的数据");
											isCompleta = false;
											// tmpdata处理完成 可以生成当日交易统计
											logger.info("tmpdata文件夹中数据全部处理完成,将进行统计信息生成");
											// 食堂进行日终操作

											logger.info("停止将buffer文件放入消息队列!");
											// Enviroment.Lock_Buffer = true;
											// Enviroment.Lock_CheckStmtDay =
											// true;
											if (DBHelper.createStatistics()) {

												logger.info("统计信息生成完毕，将对统计信息进行组包并发送至中心");
												logger.info("开始对统计信息进行组包......");
												pkg3001 = PackageHelper
														.createPackageAboutStatistics();
												if (0 == NetUtil
														.synSendMsg(pkg3001)) {
													logger.info("3001报文发送成功");
													logger.info("正在本地处理报表数据");
													if (DBHelper
															.createReportDataCNTREP()) {
														// 结算日期更改为下一天
														if (DBHelper
																.setStmtDay(strNextDay)) {
															logger.info("结算日期已设置为:"
																	+ strNextDay);
															logger.info("恢复将buffer文件放入消息队列!");
															// Enviroment.Lock_Buffer
															// = false;
															// Enviroment.Lock_CheckStmtDay
															// = false;
															logger.info("生成下一结算日的结算控制表记录成功");
														} else {
															logger.error("设置结算日期不成功");
														}
													} else {
														logger.error("处理报表数据失败");
													}
												} else {
													logger.error("发送对账信息不成功");
												}
											} else {
												logger.error("生成统计信息不成功");
											}
										} else {
											logger.info("在tmpdata文件夹内发现有  "
													+ countBuffer + " 个文件未处理！");
											logger.info("正在处理....");
											Thread.sleep(5000);
										}
									}
								} else if (days == 0) {
									logger.info("食堂与中心当前结算日一致");
									logger.info("释放消费者线程(SocketConsumer) unlock");
									SynchronizedVariable.getInstance().setBuffLock(false);
								} else if (days < 0) {
									logger.info("食堂结算日大于中心结算日");
									logger.info("释放消费者线程(SocketConsumer) unlock");
									SynchronizedVariable.getInstance().setBuffLock(false);
								}
							}
							// 判断是否中心收到食堂的日终检查报文(2002报文)
							else if (paramType.equals("2002")
									&& InitData.nodeType.equals("00")) {// 中心收到2002报文
								logger.info("中心收到2002报文  开始验证此食堂是否需要进行日终");

								// 验证2002报文长度
								if (this.pkg.length == 27) {
									// 解析食堂结算日期
									byteArrayStmtDay = new byte[4];
									System.arraycopy(this.pkg, 23,
											byteArrayStmtDay, 0, 4);
									strDrStmtDay = ByteConvert
											.bcd2Str(byteArrayStmtDay);
									logger.info("解析食堂结算日期为:" + strDrStmtDay);
									// 取中心的结算日期
									strCentreStmtDay = DBHelper.getStmtDay();
									// 转换为日期
									DrStmtDay = Convert.getDateByString(
											strDrStmtDay, "yyyyMMdd");
									CentreStmtDay = Convert.getDateByString(
											strCentreStmtDay, "yyyyMMdd");
									// 比较两个日期
									long timespan = CentreStmtDay.getTime()
											- DrStmtDay.getTime();
									days = timespan / InitData.DAY;
									if (days != 0) {
										if (days > 0) {
											logger.info("食堂节点:"
													+ this.strSupplierID
													+ "/"
													+ InitData.DR_PORT
													+ "需要进行日终操作 现在下发日终命令(2002报文)");
											// 生成食堂应切换到的日期
											strNextStmtDay = Convert
													.getSpecifiedDayAfter(strDrStmtDay);
											// 对其下发日终命令
											if (0 == NetUtil
													.synSendMsg(PackageHelper
															.createPkg2002(
																	strCentreStmtDay,
																	this.strSupplierID))) {

												logger.info("对食堂节点:"
														+ this.strSupplierID
														+ "/"
														+ InitData.DR_PORT
														+ " 下发日终报文成功!");
											} else {
												logger.info("对食堂节点:"
														+ this.strSupplierID
														+ "/"
														+ InitData.DR_PORT
														+ " 下发日终报文失败  可能无法连接此节点!");
											}
										} else {
											logger.info("食堂节点:"
													+ this.strSupplierID
													+ " 当前结算日期为:"
													+ this.strDrStmtDay
													+ " 大于中心的结算日期，不对其下发日终命令");
										}
									} else {
										logger.info("食堂节点:"
												+ this.strSupplierID
												+ " 当前结算日期与中心一致，不对其下发日终命令");
										if(0 == NetUtil
												.synSendMsg(PackageHelper
														.createPkg2003(
																strCentreStmtDay,
																this.strSupplierID))){
											logger.info("通知食堂解锁BufferReceiverServer线程");
										}
									}
								} else {
									logger.info("此2002报文长度错误，实际长度为:"
											+ this.pkg.length);
								}
							}
							// 中心收到食堂的统计报文
							else if (paramType.equals("3001")
									&& InitData.nodeType.equals("00")) {
								// 删除对账信息
								logger.info("查询食堂节点" + strSupplierID + "在"
										+ strDrStmtDay + "是否存在上传的对账报文");
								if (DBHelper
										.isExsitTransactionCompareByStmtDay(
												strSupplierID.substring(0, 2),
												strDrStmtDay)) {
									logger.info("食堂节点" + strSupplierID + "在"
											+ strDrStmtDay + "已经生成了对账报文,将进行删除");
									
									if(DBHelper.deleteTableTransactionCompareByStmtDay(
											strSupplierID.substring(0, 2),
											strDrStmtDay)){
										logger.info("食堂节点" + strSupplierID + "在"
												+ strDrStmtDay + "生成的对账报文已经删除");
										Thread.sleep(2000);
									}
								}
								else{
									logger.info("不存在食堂节点" + strSupplierID + "在"
											+ strDrStmtDay + "上传的对账报文,将进行入库");
								}
								// 将统计入库
								BufferStore.writePackageIntoFile(pkg,
										this.strSupplierID);
								logger.info("将接收到的报文存入buffer");
								logger.info("中心收到3001报文!");
								// 中心生成自己的统计报文
								// 中心检查tmpdata是否处理完成(考虑多个食堂的情况)
								// dr_tmp_path = InitData.tmpdata;
								dr_tmp_path_buffer = Convert.getDrTmpPath(
										InitData.DR_BUFFER_DATA, strSupplierID);
								dr_tmp_path_onlyread = Convert.getDrTmpPath(
										InitData.DR_ONLYREAD_DATA,
										strSupplierID);
								logger.info("将对目录" + dr_tmp_path_buffer
										+ "进行检查  检查数据是否处理完成");
								logger.info("将对目录" + dr_tmp_path_onlyread
										+ "进行检查  检查数据是否处理完成");
								fTmpdata_centre = new File(dr_tmp_path_buffer);
								fTmpdata_centre_onlyread = new File(
										dr_tmp_path_onlyread);
								boolean isCompleta = true;
								Thread.sleep(Long
										.parseLong(InitData.transmit_timeout) + 3000);
								while (isCompleta) {
									long countOnlyRead = FileUtil
											.getlist(fTmpdata_centre_onlyread);
									long countBuffer = FileUtil
											.getlist(fTmpdata_centre);
									long count = countOnlyRead + countBuffer;
									logger.info("在tmpdata文件夹中发现  " + count
											+ " 条数据未处理");
									if (count == 0) {// 表示tmpdata文件夹中无文件
														// 文件已经被处理完
										logger.info("tmpdata文件夹中没有发现未处理的数据");
										isCompleta = false;
										supplier_id = this.strSupplierID
												.substring(0, 2);
										if (DBHelper.createStatistics(
												supplier_id, strDrStmtDay)) {
											logger.info("统计信息生成完毕，将此统计信息发送给食堂节点："
													+ supplier_id);
											logger.info("开始对统计信息进行组包......");
											
											byte[] pkg3001 = PackageHelper
													.createPackageAboutStatisticsCentre(
															this.strSupplierID,
															strDrStmtDay);
											try {
												if (0 == NetUtil
														.synSendMsg(pkg3001)) {
													logger.info("3001报文已发送");
												}
											} catch (Exception e) {
											}
										} else {
											logger.error("生成统计信息不成功");
										}
									} else {
										logger.info("在tmpdata文件夹内发现有  " + count
												+ " 个文件未处理！");
										logger.info("正在处理....");
										try {
											Thread.sleep(5000);
										} catch (InterruptedException e) {
										}
									}
								}
							} else if (paramType.equals("3001")
									&& !InitData.nodeType.equals("00")) {
								// 将统计入库
								BufferStore.writePackageIntoFile(pkg);
								logger.info("将接收到的报文存入buffer");
								logger.info("食堂收到3001报文!");
								this.pkg2002_check = PackageHelper
										.createPkg2002(DBHelper.getStmtDay(),
												"00000000");
								NetUtil.synSendMsg(pkg2002_check);
							}
							// 普通报文 都只存进buffer
							else if (!paramType.equals("2001")
									&& !paramType.equals("2002")
									&& !paramType.equals("3001")
									&& InitData.nodeType.equals("00")) {
								// 将包存进buffer文件夹
								BufferStore.writePackageIntoFile(pkg,
										this.strSupplierID);

								logger.info("将接收到的报文存入buffer");
							} else if (!paramType.equals("2001")
									&& !paramType.equals("2002")
									&& !paramType.equals("3001")
									&& !InitData.nodeType.equals("00")) {
								// 将包存进buffer文件夹
								BufferStore.writePackageIntoFile(pkg);
								logger.info("将接收到的报文存入buffer");
							}
							// 重置paramType
							// paramType = "";
						} else {
							logger.error("报文头错误");
							logger.error("错误的包头为");
							logger.error(LoggerFanka.dumpBin(head));
						}
					} else {
						logger.info("计算要接收的包长度错误 包长度小于23字节!");
					}
				} catch (DataTranslateException e) {
					logger.info(e.getMessage());
					continue;
				} catch (InterruptedException e) {
				}
			} catch (IOException e) {
				logger.error("serverSocket.accept()失败");
				logger.error(e.getMessage());
				continue;
			} finally {
				try {
					socket.close();
					if (in != null) {
						in.close();
					}
					if (out != null) {
						out.close();
					}
					pkg = null;

				} catch (IOException e) {
					logger.error("释放socket失败");
					logger.error(e.getMessage());
				}
			}
		}
	}

	// 23长包头
	private byte getResponse(byte[] pkg) {
		byte ask = 0x00;
		int offset = 0;
		int pkgLength = pkg.length;
		int oneRecordLength = 0;
		if (pkgLength > 23) {
			head = new byte[23];
			System.arraycopy(pkg, 0, head, 0, 23);
			byte[] paramTypeByte = new byte[2];
			System.arraycopy(head, offset, paramTypeByte, 0, 2);
			offset += 2;
			paramType = ByteConvert.bcd2Str(paramTypeByte);
			if (paramType.length() == 4) {
				if(paramType.equals("2003")){
					return ask;
				}
				else{
					oneRecordLength = FileUtil.isExistWithParamType(paramType);
					logger.info("包头解析 " + paramType + "参数的单条记录长度为:"
							+ oneRecordLength);
					if (oneRecordLength != 0) {
						try {
							sender = new byte[4];
							System.arraycopy(head, offset, sender, 0, 4);
							offset += 4;
							strSupplierID = ByteConvert.bcd2Str(sender);
							tmpMap = InitData.getInstance().getSupplierMap()
									.get(strSupplierID);
							if (tmpMap != null) {
								strSUpplierIP = tmpMap.split(":")[0];
								logger.info("包头解析-->解析报文的发送方为:" + strSupplierID
										+ " IP为:" + strSUpplierIP);
								receiver = new byte[4];
								System.arraycopy(head, offset, receiver, 0, 4);
								strReceiverID = ByteConvert.bcd2Str(receiver);
								logger.info("包头解析-->接收方节点ID:" + strReceiverID);
								strReceiverIP = InitData.getInstance()
										.getSupplierMap().get(strReceiverID)
										.split(":")[0];
								logger.info("包头解析-->解析报文的接收方为:" + strReceiverID
										+ " IP为:" + strReceiverIP);
								offset += 4;
								/*
								 * portByte = new byte[2]; System.arraycopy(head,
								 * offset, portByte, 0, 2); port =
								 * ByteConvert.byteToShort(portByte);
								 */
								port = Short.parseShort(InitData.getInstance()
										.getSupplierMap().get(strReceiverID)
										.split(":")[1]);
								offset += 2;
								msgMgrType = new byte[1];
								System.arraycopy(head, offset, msgMgrType, 0, 1);
								offset += 1;
								if (msgMgrType[0] == 0x00 || msgMgrType[0] == 0x01
										|| msgMgrType[0] == 0x02
										|| msgMgrType[0] == 0x03) {

								} else {
									logger.error("未知的消息处理类型:" + msgMgrType[0]);
									ask = Ask.EXPECT_NODE_TYPE;
									return ask;
								}
								seq = new byte[4];
								System.arraycopy(head, offset, seq, 0, 4);
								offset += 4;
								int iSeq = ByteConvert.byteToInt(seq);
								pkgno = new byte[2];
								System.arraycopy(head, offset, pkgno, 0, 2);
								offset += 2;
								short pkgNo = ByteConvert.byteToShort(pkgno);
								flag = new byte[1];
								System.arraycopy(head, offset, flag, 0, 1);
								offset += 1;
								recordcount = new byte[2];
								System.arraycopy(head, offset, recordcount, 0, 2);
								offset += 2;
								short recordCount = ByteConvert
										.byteToShort(recordcount);
								int tmpLength = (pkg.length - headLength)
										/ oneRecordLength;
								logger.info("计算当前解析包的记录数为:" + tmpLength);
								logger.info("包头中解析到的包记录数为:" + recordCount);
								if (recordCount != tmpLength) {
									logger.error("包记录数与包长度不匹配");
									ask = Ask.RECORD_COUNT_CANT_MATCHING_PACKAGE_LENGTH;
								}
								agm = new byte[1];
								System.arraycopy(head, offset, agm, 0, 1);
								/* *
								 * test
								 */

								logger.info("参数类型为:" + paramType);
								logger.info("发送方标识:" + strSupplierID);
								logger.info("接收方标识:" + strReceiverID);
								logger.info("接收方端口:" + port);
								logger.info("消息处理类型:" + msgMgrType[0]);
								logger.info("流水号:" + iSeq);
								logger.info("包编号:" + pkgNo);
								logger.info("包标识:" + flag[0]);
								logger.info("包记录数:" + recordCount);
							} else {
								logger.info("接受到的发送节点为:" + strSupplierID
										+ " 未找到该节点对应的IP和端口");
							}

						} catch (DataTranslateException e) {
							logger.error("数据类型转换错误");
							logger.error(e.getMessage(), e);
							ask = Ask.EXPECT_NODE_TYPE;
						} finally {
							pkg = null;
							msgTypeByte = null;
							sender = null;
							seq = null;
							pkgno = null;
							flag = null;
							recordcount = null;
							agm = null;
						}
					} else {
						ask = Ask.INVALID_PARAM_TYPE;
					}
				}
			} else {
				ask = Ask.INVALID_PARAM_TYPE;
			}
		} else {
			ask = Ask.ERROR_PKG;
			logger.info("该报文包体为空 或 为止报文");
		}
		return ask;
	}
}
